package com.alibaba.otter.manager.web.service;

import com.alibaba.otter.manager.biz.config.channel.ChannelService;
import com.alibaba.otter.manager.biz.statistics.throughput.ThroughputStatService;
import com.alibaba.otter.manager.biz.statistics.throughput.param.*;
import com.alibaba.otter.manager.web.dto.HistoryThroughputInfoDTO;
import com.alibaba.otter.manager.web.dto.RecentThroughputStatDTO;
import com.alibaba.otter.manager.web.mapstruct.ThroughputMapper;
import com.alibaba.otter.shared.common.model.config.channel.Channel;
import com.alibaba.otter.shared.common.model.statistics.throughput.ThroughputStat;
import com.alibaba.otter.shared.common.model.statistics.throughput.ThroughputType;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

@Service
public class ThroughputStatAggService {


    private final ThroughputStatService throughputStatService;

    private final ThroughputMapper throughputMapper;

    private final ChannelService channelService;

    public ThroughputStatAggService(ThroughputStatService throughputStatService,
                                    ThroughputMapper throughputMapper,
                                    ChannelService channelService) {
        this.throughputStatService = throughputStatService;
        this.throughputMapper = throughputMapper;
        this.channelService = channelService;
    }


    public List<RecentThroughputStatDTO> findRecentThroughputStatByPipelineId(Long pipelineId) {

        List<RecentThroughputStatDTO> recentThroughputStats = new ArrayList<>();

        List<AnalysisType> analysisType = new ArrayList<>();
        analysisType.add(AnalysisType.ONE_MINUTE);
        analysisType.add(AnalysisType.FIVE_MINUTE);
        analysisType.add(AnalysisType.FIFTEEN_MINUTE);

        ThroughputCondition fileThroughputStatQuery = new ThroughputCondition();
        fileThroughputStatQuery.setPipelineId(pipelineId);
        fileThroughputStatQuery.setType(ThroughputType.FILE);
        ThroughputStat fileThroughputStat = throughputStatService.findThroughputStatByPipelineId(fileThroughputStatQuery);

        if (fileThroughputStat.getPipelineId() != null) {
            RealtimeThroughputCondition fileThroughputInfoQuery = new RealtimeThroughputCondition();
            fileThroughputInfoQuery.setPipelineId(pipelineId);
            fileThroughputInfoQuery.setAnalysisType(analysisType);
            fileThroughputInfoQuery.setType(ThroughputType.FILE);
            Map<AnalysisType, ThroughputInfo> fileThroughputInfo = throughputStatService.listRealtimeThroughput(fileThroughputInfoQuery);
            RecentThroughputStatDTO fileRecentThroughputStatDTO = throughputMapper.toRecentThroughputStatDTO(fileThroughputStat, fileThroughputInfo);
            recentThroughputStats.add(fileRecentThroughputStatDTO);
        }

        ThroughputCondition rowThroughputStatQuery = new ThroughputCondition();
        rowThroughputStatQuery.setPipelineId(pipelineId);
        rowThroughputStatQuery.setType(ThroughputType.ROW);

        ThroughputStat rowThroughputStat = throughputStatService.findThroughputStatByPipelineId(rowThroughputStatQuery);
        if (rowThroughputStat.getPipelineId() != null) {
            RealtimeThroughputCondition rowThroughputInfoQuery = new RealtimeThroughputCondition();
            rowThroughputInfoQuery.setPipelineId(pipelineId);
            rowThroughputInfoQuery.setAnalysisType(analysisType);
            rowThroughputInfoQuery.setType(ThroughputType.ROW);
            Map<AnalysisType, ThroughputInfo> rowThroughputInfo = throughputStatService.listRealtimeThroughput(rowThroughputInfoQuery);
            RecentThroughputStatDTO rowRecentThroughputStatDTO = throughputMapper.toRecentThroughputStatDTO(rowThroughputStat, rowThroughputInfo);
            recentThroughputStats.add(rowRecentThroughputStatDTO);
        }


        return recentThroughputStats;
    }


    public HistoryThroughputInfoDTO historyThroughput(Instant startTime, Instant endTime, Long pipelineId) {
        if (startTime == null || endTime == null) {
            startTime = Instant.ofEpochMilli(System.currentTimeMillis() / 60000 * 60000 - 24 * 60 * 60 * 1000);
            endTime = Instant.ofEpochMilli(System.currentTimeMillis() / 60000 * 60000);
        }

        Channel channel = channelService.findByPipelineId(pipelineId);
        TimelineThroughputCondition rowThroughputCond = new TimelineThroughputCondition();
        TimelineThroughputCondition fileThroughputCond = new TimelineThroughputCondition();

        rowThroughputCond.setStart(startTime);
        rowThroughputCond.setEnd(endTime);
        rowThroughputCond.setType(ThroughputType.ROW);
        rowThroughputCond.setPipelineId(pipelineId);
        fileThroughputCond.setStart(startTime);
        fileThroughputCond.setEnd(endTime);
        fileThroughputCond.setType(ThroughputType.FILE);
        fileThroughputCond.setPipelineId(pipelineId);
        Map<Long, ThroughputInfo> rowThroughputInfo = throughputStatService.listTimelineThroughput(rowThroughputCond);
        Map<Long, ThroughputInfo> fileThroughputInfo = throughputStatService.listTimelineThroughput(fileThroughputCond);


        Long totalRowRecord = 0L;
        Long totalFileRecord = 0L;
        Long totalRowSize = 0L;
        Long totalFileSize = 0L;
        for (ThroughputInfo info : rowThroughputInfo.values()) {
            totalRowRecord += info.getNumber();
            totalRowSize += info.getSize();
        }

        for (ThroughputInfo info : fileThroughputInfo.values()) {
            totalFileRecord += info.getNumber();
            totalFileSize += info.getSize();
        }

        Map<Long, RecentThroughputStatDTO.ThroughputInfoDTO> rowThroughputStatDTO = throughputMapper.toMapThroughputStatDTO(rowThroughputInfo);
        Map<Long, RecentThroughputStatDTO.ThroughputInfoDTO> fileThroughputStatDTO = throughputMapper.toMapThroughputStatDTO(fileThroughputInfo);
        HistoryThroughputInfoDTO historyThroughput = new HistoryThroughputInfoDTO();
        historyThroughput.setStart(startTime);
        historyThroughput.setEnd(endTime);
        historyThroughput.setPipelineId(pipelineId);
        historyThroughput.setChannel(channel);
        historyThroughput.setTotalRowRecord(totalRowRecord);
        historyThroughput.setTotalFileRecord(totalFileRecord);
        historyThroughput.setTotalRowSize(totalRowSize);
        historyThroughput.setTotalFileSize(totalFileSize);
        historyThroughput.setRowThroughput(rowThroughputStatDTO);
        historyThroughput.setFileThroughput(fileThroughputStatDTO);

        return historyThroughput;
    }
}
